package cn.caplike.demo.canal;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import org.springframework.lang.NonNull;

import java.net.InetSocketAddress;
import java.text.MessageFormat;
import java.util.List;

/**
 * Description: Canal client under tcp mode.
 *
 * @author <a href="mailto:like@pystandard.com">LiKe</a> @ 2023-10-26 14:41:20
 */
public class CanalClient {
    // @formatter:off
    public static final String CANAL_INSTANCE = "10.20.100.40";
    public static final String CANAL_DEPLOYER_HOSTNAME = "big-data-2";
    public static final int    CANAL_DEPLOYER_PORT = 11111;

    public static final String MYSQL_DB_TO_SUBSCRIBE = "sync.*";
    // @formatter:on

    @SuppressWarnings({"InfiniteLoopStatement", "BusyWait"})
    public static void main(String[] args) {
        // 获取连接
        final CanalConnector connector = connect();
        connector.subscribe(MYSQL_DB_TO_SUBSCRIBE); // 订阅数据库

        while (true) {
            final Message message = connector.get(100); // 一次 Canal 从日志中抓取的信息, 一个 Message 中可以包含多个 SQL 执行的结果
            final List<CanalEntry.Entry> entries = message.getEntries();

            if (entries.isEmpty()) {
                System.out.println("No data fetched, sleep 5 secs ...");
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException ignored) {
                }
            } else {
                for (final CanalEntry.Entry entry : entries) {
                    final String tableName = entry.getHeader().getTableName();

                    final CanalEntry.EntryType entryType = entry.getEntryType();
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
                        try {
                            final ByteString storeValue = entry.getStoreValue();

                            // 反序列化数据
                            final CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

                            // 获取当前事件的操作类型
                            final CanalEntry.EventType eventTypeOfRowChange = rowChange.getEventType();

                            // 获取数据集
                            final List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();

                            for (CanalEntry.RowData rowData : rowDatasList) {
                                final JSONObject before = new JSONObject();
                                for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                                    before.put(column.getName(), column.getValue());
                                }

                                final JSONObject after = new JSONObject();
                                for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
                                    after.put(column.getName(), column.getValue());
                                }

                                System.out.println("Sql: " + rowChange.getSql()); // todo: 需要 binlog 模式为 statement?
                                System.out.println(
                                        MessageFormat.format(
                                                "Table: {0} [{1}] \r\n\t- before: {2}\r\n\t- after: {3}",
                                                tableName, eventTypeOfRowChange, before, after));
                            }
                        } catch (InvalidProtocolBufferException e) {
                            throw new RuntimeException(e);
                        }
                    } else {
                        System.out.println("Unhandled entry type:  " + entryType);
                    }
                }
            }
        }
    }

    private static @NonNull CanalConnector connect() {
        final InetSocketAddress canalAddress = new InetSocketAddress(CANAL_DEPLOYER_HOSTNAME, CANAL_DEPLOYER_PORT);
        // Canal 的 username 和 password 并没有
        final CanalConnector connector = CanalConnectors.newSingleConnector(canalAddress, CANAL_INSTANCE, "", "");
        connector.connect();
        return connector;
    }
}
